package com.yyx.demo.publishSubscribe;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.TreeMap;


//内部操作类,顺序存储，始终保留前五个消费信息，超过就删
public class SequenceZnode {
    private Long currentId=-1L;

    private final Logger LOG = LoggerFactory.getLogger(SequenceZnode.class);

    private ZooKeeper zookeeper;
    private List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;

    private final int DELETE_LOWEST_LIMIT=5;


    private final String prefix = "qn-";

    public SequenceZnode(ZooKeeper zookeeper, List<ACL> acl) {
        if (acl != null) {
            this.acl = acl;
        }
        this.zookeeper = zookeeper;
    }

    //同步提交
    public boolean offer(byte[] data,String dir) throws KeeperException, InterruptedException {
        while (true) {
            try {
                zookeeper.create(dir + "/" + prefix, data,acl, CreateMode.PERSISTENT_SEQUENTIAL);
                return true;
                //节点不存在则创建节点
            } catch (KeeperException.NoNodeException e) {
                zookeeper.create(dir, new byte[0], acl, CreateMode.PERSISTENT);
            }
        }
    }

    public byte[] take(String dir) throws KeeperException, InterruptedException {
        Map<Long, String> orderedChildren;
        // Same as for element.  Should refactor this.
        while (true) {

            try {
                orderedChildren = orderedChildren(dir);
            } catch (KeeperException.NoNodeException e) {
                zookeeper.create(dir, new byte[0], acl, CreateMode.PERSISTENT);
                continue;
            }
            if (orderedChildren.size() == 0) {
                return null;
            }

            String path;
            if(currentId==-1L){
                for (String headNode : orderedChildren.values()) {
                    path = dir + "/" + headNode;
                    try {
                        byte[] data = zookeeper.getData(path, false, null);
                        currentId =  Long.parseLong(headNode.substring(prefix.length()))+1;
                        return data;
                    } catch (KeeperException.NoNodeException e) {
                        // Another client deleted the node first.
                    }
                }
            }else {
                String currentNode=orderedChildren.get(currentId);
                path = dir + "/" + currentNode;
                byte[] data = zookeeper.getData(path, false, null);
                currentId =  Long.parseLong(currentNode.substring(prefix.length()))+1;
                return data;
            }
        }
    }

    //以id构成的结构体，存元
    private Map<Long, String> orderedChildren(String dir) throws KeeperException, InterruptedException {
        Map<Long, String> orderedChildren = new TreeMap<>();
        List<String> childNames;
        childNames = zookeeper.getChildren(dir, null);

        for (String childName : childNames) {
            try {
                //检查名字是否一样,避免错误ZK操作
                if (!childName.regionMatches(0, prefix, 0, prefix.length())) {
                    LOG.warn("Found child node with improper name: {}", childName);
                    continue;
                }
                String suffix = childName.substring(prefix.length());
                Long childId = Long.parseLong(suffix);

                Long lower = currentId-childId;
                if(lower>=DELETE_LOWEST_LIMIT){
                    //删除已经过期消费的节点，保留5个
                    zookeeper.delete("/"+childName,-1);
                }else {
                    orderedChildren.put(childId, childName);
                }
            } catch (NumberFormatException e) {
                LOG.warn("Found child node with improper format : {}", childName, e);
            }
        }
        return orderedChildren;
    }

}
